package com.xy.kafka.consumer;

import com.xy.kafka.service.SparkStreamingService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumerListener {

    private final SparkStreamingService sparkStreamingService;

    @Autowired
    public KafkaConsumerListener(SparkStreamingService sparkStreamingService) {
        this.sparkStreamingService = sparkStreamingService;
    }

    /*
     * 监听tc-topic主题的消息并传递给SparkStreamingService处理
     * @param record 消息记录
     */
    @KafkaListener(topics = "tc-topic", groupId = "test-group")
    public void listenTcTopic(ConsumerRecord<String, String> record) {
        String message = record.value();
        System.out.println("收到消息: 分区=" + record.partition() + 
                          ", offset=" + record.offset() + 
                          ", 消息内容=" + message);
        
        // 将消息传递给SparkStreamingService进行处理
        try {
            sparkStreamingService.processMessage(message);
        } catch (Exception e) {
            System.err.println("处理消息时出错: " + e.getMessage());
        }
    }
}